package com.alibaba.nacos.core.remote.grpc;

import com.alibaba.nacos.core.remote.BaseRpcServer;
import com.alibaba.nacos.core.remote.ConnectionManager;
import com.google.common.util.concurrent.MoreExecutors;
import io.grpc.*;
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
import io.grpc.util.MutableHandlerRegistry;
import lombok.Getter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

@Getter
@Slf4j
public abstract class BaseGrpcServer
        extends BaseRpcServer {

    @Autowired
    private GrpcRequestAcceptor grpcCommonRequestAcceptor;
    @Autowired
    private GrpcBiStreamRequestAcceptor grpcBiStreamRequestAcceptor;
    @Autowired
    private ConnectionManager connectionManager;

    private Server server;

    private int port = 8858;
    private int maxInboundMessageSize = 10 * 1024 * 1024;
    private long keepAliveTime = TimeUnit.HOURS.toMillis(2);
    private long keepAliveTimeout = TimeUnit.SECONDS.toMillis(20);
    private long permitKeepAliveTime = TimeUnit.MINUTES.toMillis(5);
    private Executor executor = MoreExecutors.directExecutor();

    @SneakyThrows
    @Override
    protected void startServer() {
        log.info("start server on port {}", port + rpcPortOffset());
        NettyServerBuilder builder = NettyServerBuilder.forPort(port + rpcPortOffset()).executor(executor);

        MutableHandlerRegistry handlerRegistry = new MutableHandlerRegistry();
        ServerInterceptor serverInterceptor = new GrpcConnectionInterceptor();
        handlerRegistry.addService(ServerInterceptors.intercept(grpcCommonRequestAcceptor, serverInterceptor));
        handlerRegistry.addService(ServerInterceptors.intercept(grpcBiStreamRequestAcceptor, serverInterceptor));

        server = builder.maxInboundMessageSize(getMaxInboundMessageSize())
                .fallbackHandlerRegistry(handlerRegistry)
                .addTransportFilter(new AddressTransportFilter(connectionManager))
                .compressorRegistry(CompressorRegistry.getDefaultInstance())
                .decompressorRegistry(DecompressorRegistry.getDefaultInstance())
                .keepAliveTime(getKeepAliveTime(), TimeUnit.MILLISECONDS)
                .keepAliveTimeout(getKeepAliveTimeout(), TimeUnit.MILLISECONDS)
                .permitKeepAliveTime(getPermitKeepAliveTime(), TimeUnit.MILLISECONDS)
                .build();
        server.start();

        log.info("started server on port {}", port + rpcPortOffset());
    }
}
